package com.golxen.framework.netty.session.base;


import com.golxen.framework.PBPacket.PBPacket;
import com.golxen.utils.LogUtil;
import io.netty.channel.Channel;
import io.netty.util.concurrent.ScheduledFuture;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class NettyChannelSession extends AbstractSession {

    private final Channel channel;

    private final ConcurrentLinkedQueue<PBPacket> bufferQueue = new ConcurrentLinkedQueue<>();

    private final AtomicLong counter = new AtomicLong(0);

    private int linkIndex;

    ScheduledFuture<?> scheduleWithFixedDelay;

    public NettyChannelSession(Channel channel, int linkIndex) {
        this.linkIndex = linkIndex;
        this.channel = channel;

        scheduleWithFixedDelay = channel.eventLoop().scheduleWithFixedDelay(() -> {

            if (!channel.isActive()) { // 如果channel断了，取消定时
                scheduleWithFixedDelay.cancel(false);
                return;
            }

            int i = 0;
            // 若有消息要推，且写缓存池未满

            int cause = 0;

            for (;;) {
                if (bufferQueue.isEmpty()) {
                    cause = 1;
                    break;
                }
                if (!channel.isWritable()) {
                    cause = 2;
                    break;
                }
                PBPacket poll = bufferQueue.poll();
                if (poll == null) {
                    cause = 3;
                    break;
                }
                counter.decrementAndGet();
                channel.writeAndFlush(poll);
                i++;
            }

            long left = counter.get();

            if (i == 0 && cause == 1) {
                return;
            }

            if (left < 10000) {
                return;
            }

            LogUtil.COMMON.info("{} write and flush {} break cause {} left {}", //
                    channel == null ? "":channel.id().asShortText(), //
                    i, //
                    cause, //
                    left //
            );

        }, 1, 2, TimeUnit.MILLISECONDS);
    }

    public int getLinkIndex() {
        return linkIndex;
    }

    @Override
    public void write(PBPacket msg) {
        if (msg == null) {
            return;
        }
        counter.incrementAndGet();
        bufferQueue.offer(msg);
    }

    public void send(PBPacket msg) {
        write(msg);
    }

    public void send(int playerId, PBPacket msg) {
        msg.setPlayerId(playerId);
        write(msg);
    }

    @Override
    public boolean isAlive() {
        try {
            if (channel == null) {
                return false;
            }
            return channel.isActive();
        } catch (Exception e) {
            LogUtil.COMMON.error("channel is not active,error:", e);
            return false;
        }
    }

    @Override
    public void close() {
        channel.close();
    }

    public Channel getChannel() {
        return channel;
    }
}
